package com.atguigu.chapter05.source;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.Socket;
import java.nio.charset.StandardCharsets;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/9 10:37
 */
public class Flink04_Source_Custom {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        env.addSource(new MySocketSource("hadoop162", 9999)).print();
        
        env.execute();
    }
    
    static class MySocketSource implements SourceFunction<WaterSensor> {
        
        private String host;
        private int port;
        private boolean isCancel;
    
        public MySocketSource(String host, int port) {
            this.host = host;
            this.port = port;
        }
        
        // 生产数据, 把数据写入到流中
        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            // socket 编程
            Socket socket = new Socket(host, port);
            BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8));
            String line = reader.readLine();
            while (line != null && !isCancel) {
                String[] data = line.split(",");
                ctx.collect(new WaterSensor(data[0], Long.valueOf(data[1]), Integer.valueOf(data[2])));
                
                line = reader.readLine();
                
            }
        }
        
        // 停止source, 不再产生数据
        @Override
        public void cancel() {
            isCancel =true;
        }
    }
}
